package io.kiki.sba.registry.server.metadata.slot.manager;

import com.google.common.collect.ImmutableMap;
import io.kiki.sba.registry.common.model.slot.DataStoreNode;
import io.kiki.sba.registry.common.model.slot.SlotConfig;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.observer.impl.AbstractLifecycleObservable;
import io.kiki.sba.registry.server.metadata.slot.SlotManager;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

@Setter
@Getter
public class SlotManagerImpl extends AbstractLifecycleObservable implements SlotManager {
    private static final Logger logger = LoggerFactory.getLogger(SlotManagerImpl.class);
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    private volatile SlotMapWrapper slotMapWrapper = new SlotMapWrapper(Slots.INIT, ImmutableMap.of());

    @Setter
    private int slotNums = SlotConfig.SLOT_NUM;
    @Setter
    private int slotReplicas = SlotConfig.SLOT_REPLICAS;

    @Override
    public Slots getSlots() {
        lock.readLock().lock();
        try {
            return slotMapWrapper.getSlots();
        } finally {
            lock.readLock().unlock();
        }
    }

    /**
     * The function means to update slot-table if possible(like CAS(xxx, xxx)) return 1. true: if
     * there were any changes on slot-table (by check whether slot-table epoch getting bigger) 2.
     * false: if there were no changes, including we try to refresh a lower or equal epoch slot-table
     *
     * <p>Then, it will notify observers about this slot-table change event
     */
    @Override
    public boolean refresh(Slots slots) {
        lock.writeLock().lock();
        try {
            long localEpoch = slotMapWrapper.getSlots().getEpoch();
            // same epoch means no changes here, so we try not to notify observers
            if (slots.getEpoch() == localEpoch) {
                return false;
            }
            if (slots.getEpoch() < localEpoch) {
                if (logger.isWarnEnabled()) {
                    logger.warn("[refresh]receive slot table,but epoch({}) is smaller than current({})", slots.getEpoch(), slotMapWrapper.getSlots().getEpoch());
                }
                return false;
            }
            setSlotMapWrapper(new SlotMapWrapper(slots, refreshReverseMap(slots)));
        } finally {
            lock.writeLock().unlock();
        }
        return true;
    }

    private Map<String, DataStoreNode> refreshReverseMap(Slots slots) {
        Map<String, DataStoreNode> map = new HashMap<>();
        List<DataStoreNode> dataStoreNodes = slots.transfer(null, false);
        for (DataStoreNode dataStoreNode : dataStoreNodes) {
            map.put(dataStoreNode.getDataStoreNodeId(), dataStoreNode);
        }
        return ImmutableMap.copyOf(map);
    }

    @Override
    public int getSlotNums() {
        return slotNums;
    }


    @Override
    public int getSlotReplicaNums() {
        return slotReplicas;
    }


    @Override
    public DataStoreNode getDataNodeManagedSlot(String dataNode, boolean ignoreFollowers) {
        lock.readLock().lock();
        try {
            // here we ignore port for data-node, as when store the reverse-map, we lose the port
            // information
            // besides, port is not matter here
            DataStoreNode dataStoreNode = slotMapWrapper.getReverseMap().get(dataNode);
            if (dataStoreNode == null) {
                return new DataStoreNode(dataNode);
            }
            return dataStoreNode.fork(ignoreFollowers);
        } finally {
            lock.readLock().unlock();
        }
    }

}
